package cn.lokei.gateway.handler.websocket;

import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;

import io.netty.buffer.ByteBuf;
import io.netty.buffer.Unpooled;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelFutureListener;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.SimpleChannelInboundHandler;
import io.netty.channel.group.ChannelGroup;
import io.netty.channel.group.DefaultChannelGroup;
import io.netty.handler.codec.http.DefaultFullHttpResponse;
import io.netty.handler.codec.http.FullHttpRequest;
import io.netty.handler.codec.http.HttpResponseStatus;
import io.netty.handler.codec.http.HttpVersion;
import io.netty.handler.codec.http.websocketx.CloseWebSocketFrame;
import io.netty.handler.codec.http.websocketx.PingWebSocketFrame;
import io.netty.handler.codec.http.websocketx.PongWebSocketFrame;
import io.netty.handler.codec.http.websocketx.WebSocketFrame;
import io.netty.handler.codec.http.websocketx.WebSocketServerHandshaker;
import io.netty.handler.codec.http.websocketx.WebSocketServerHandshakerFactory;
import io.netty.util.CharsetUtil;
import io.netty.util.concurrent.GlobalEventExecutor;

import static io.netty.handler.codec.http.HttpUtil.isKeepAlive;

public class WebsocketMqttHandler extends SimpleChannelInboundHandler<Object> {
  // private final Logger logger = Logger.getLogger(this.getClass());
  private static Logger logger = LogManager.getLogger(WebsocketMqttHandler.class.getName());

  // 用于记录和管理所有客户端的channle
  private static ChannelGroup clients = new DefaultChannelGroup(GlobalEventExecutor.INSTANCE);

  private WebSocketServerHandshaker handshaker;

  @Override
  public void channelRegistered(ChannelHandlerContext ctx) throws Exception {
    super.channelRegistered(ctx);
  }

  // @Override
  // public void channelRead(ChannelHandlerContext ctx, Object msg) throws
  // Exception, IOException {
  // if (msg instanceof MqttMessage) {
  // return;
  // }
  // }

  @Override
  protected void channelRead0(ChannelHandlerContext ctx, Object msg) throws Exception {
    // 获取客户端传输过来的消息
    // String content = msg.text();
    // System.out.println("接受到的数据：" + content);

    // // for (Channel channel: clients) {
    // // channel.writeAndFlush(
    // // new TextWebSocketFrame(
    // // "[服务器在]" + LocalDateTime.now()
    // // + "接受到消息, 消息为：" + content));
    // // }
    // // 下面这个方法，和上面的for循环，一致 向客户端发送数据
    // clients.writeAndFlush(new TextWebSocketFrame("我是服务器,我收到你的消息为:" + content));
    logger.info("收到消息：" + msg);
    if (msg instanceof FullHttpRequest) {
      // 以http请求形式接入，但是走的是websocket
      handleHttpRequest(ctx, (FullHttpRequest) msg);
    } else if (msg instanceof WebSocketFrame) {
      // 处理websocket客户端的消息
      handlerWebSocketFrame(ctx, (WebSocketFrame) msg);
    }

  }

  /**
   * 当客户端连接服务端之后（打开连接） 获取客户端的channle，并且放到ChannelGroup中去进行管理
   */
  @Override
  public void handlerAdded(ChannelHandlerContext ctx) throws Exception {
    clients.add(ctx.channel());
    ctx.writeAndFlush(Unpooled.copiedBuffer("aaaa".getBytes()));
  }

  @Override
  public void channelActive(ChannelHandlerContext ctx) throws Exception {
    System.out.println("客户端传给服务端");
    String string = "116.478935,39.997761";
    // ByteBuf buf = Unpooled.buffer(string.length());
    // buf.retain();
    // buf.writeBytes(string.getBytes());
    // ctx.writeAndFlush(buf);
    //
    // System.out.println(ctx.writeAndFlush(b\
    // ctx.channel().writeAndFlush(new TextWebSocketFrame(string));
    System.out.println(string);
    super.channelActive(ctx);
  }

  @Override
  public void handlerRemoved(ChannelHandlerContext ctx) throws Exception {
    // 当触发handlerRemoved，ChannelGroup会自动移除对应客户端的channel,所以下面的remove不用我们再手写
    // clients.remove(ctx.channel());
    System.out.println("客户端断开，channle对应的长id为：" + ctx.channel().id().asLongText());
    System.out.println("客户端断开，channle对应的短id为：" + ctx.channel().id().asShortText());
  }

  private void handlerWebSocketFrame(ChannelHandlerContext ctx, WebSocketFrame frame) {
    // 判断是否关闭链路的指令
    if (frame instanceof CloseWebSocketFrame) {
      handshaker.close(ctx.channel(), (CloseWebSocketFrame) frame.retain());
      return;
    }
    // 判断是否ping消息
    if (frame instanceof PingWebSocketFrame) {
      ctx.channel().write(new PongWebSocketFrame(frame.content().retain()));
      return;
    }
    ByteBuf echoMsg = frame.content();
    frame.retain();
    ctx.fireChannelRead(echoMsg);
    // 本例程仅支持文本消息，不支持二进制消息
    // if (!(frame instanceof TextWebSocketFrame)) {
    // logger.debug("本例程仅支持文本消息，不支持二进制消息");
    // throw new UnsupportedOperationException(
    // String.format("%s frame types not supported", frame.getClass().getName()));
    // }
    // // 返回应答消息
    // String request = ((TextWebSocketFrame) frame).text();
    // logger.info("服务端收到：" + request);
    // TextWebSocketFrame tws = new TextWebSocketFrame(new Date().toString() +
    // ctx.channel().id() + "：" + request);
    // // 群发
    // ChannelSupervise.send2All(tws);
    // 返回【谁发的发给谁】
    // ctx.channel().writeAndFlush(tws);
  }

  /**
   * 唯一的一次http请求，用于创建websocket
   */
  private void handleHttpRequest(ChannelHandlerContext ctx, FullHttpRequest req) {
    // 要求Upgrade为websocket，过滤掉get/Post
    if (!req.decoderResult().isSuccess() || (!"websocket".equals(req.headers().get("Upgrade")))) {
      // 若不是websocket方式，则创建BAD_REQUEST的req，返回给客户端
      sendHttpResponse(ctx, req, new DefaultFullHttpResponse(HttpVersion.HTTP_1_1, HttpResponseStatus.BAD_REQUEST));
      return;
    }
    logger.info(req.content());
    // ctx.fireChannelRead(req.content());
    WebSocketServerHandshakerFactory wsFactory = new WebSocketServerHandshakerFactory("ws://localhost:1885/websocket",
        "mqtt", false);
    handshaker = wsFactory.newHandshaker(req);
    if (handshaker == null) {
      WebSocketServerHandshakerFactory.sendUnsupportedVersionResponse(ctx.channel());
    } else {
      req.headers().set("Sec-WebSocket-Protocol", "mqtt");
      handshaker.handshake(ctx.channel(), req);
    }
  }

  /**
   * 拒绝不合法的请求，并返回错误信息
   */
  private static void sendHttpResponse(ChannelHandlerContext ctx, FullHttpRequest req, DefaultFullHttpResponse res) {
    // 返回应答给客户端
    if (res.status().code() != 200) {
      ByteBuf buf = Unpooled.copiedBuffer(res.status().toString(), CharsetUtil.UTF_8);
      res.content().writeBytes(buf);
      buf.release();
    }
    ChannelFuture f = ctx.channel().writeAndFlush(res);
    // 如果是非Keep-Alive，关闭连接
    if (!isKeepAlive(req) || res.status().code() != 200) {
      f.addListener(ChannelFutureListener.CLOSE);
    }
  }
}
